package com.bigdata.warehouse.dws;

import com.bigdata.warehouse.entity.CarBehavior;
import com.bigdata.warehouse.sink.DwsCarBehaviorClickhouseSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

public class DwsCarBehavior {
    public static void main(String[] args) throws Exception {
        //1.获取Stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        //2.创建表的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

        //开启checkpoint容错
//        senv.enableCheckpointing(60000);
//        senv.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        senv.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
//        senv.getCheckpointConfig().setCheckpointTimeout(10000);
//        senv.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        //设置状态后端
        //（1）开启RocksDB
//        senv.setStateBackend(new EmbeddedRocksDBStateBackend());
//        //（2）设置checkpoint存储位置
//        senv.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://mycluster/flink/checkpoints"));

        //3.读取dwd层车辆明细表
        tEnv.executeSql("CREATE TABLE dwd_cars_log (\n" +
                " id STRING,\n" +
                " owerId INT,\n" +
                " opTime STRING,\n" +
                " cId BIGINT,\n" +
                " carCode STRING,\n" +
                " ctype SMALLINT,\n" +
                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(opTime),0),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '1' SECOND"+
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'dwd_cars_log',\n" +
                " 'properties.bootstrap.servers' = 'master:9092',\n" +
                " 'properties.group.id' = 'dwd_cars_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")\n");

        //4.读取用户维表
        tEnv.executeSql("CREATE TABLE dim_base_ower(\n" +
                " id INT,\n" +
                " roomId INT,\n" +
                " owerCode STRING,\n" +
                " owerName STRING,\n" +
                " owerSex STRING,\n" +
                " owerAge INT,\n" +
                " owerIdcard STRING,\n" +
                " owerTel STRING,\n" +
                " owerHomeplace STRING,\n" +
                " owerWorkplace STRING,\n" +
                " remark STRING,\n" +
                " createTime TIMESTAMP,\n" +
                " operatorTime TIMESTAMP,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ")  WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_ower',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");
        //5.读取车辆维度表
        tEnv.executeSql("CREATE TABLE dim_base_cars(\n" +
                " id INT,\n" +
                " owerId INT,\n" +
                " carCode STRING,\n" +
                " carColor STRING,\n" +
                " type TINYINT,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY(id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive',\n" +
                " 'table-name' = 'dim_base_cars'\n" +
                ")\n");
        //6.读取社区维度表
        tEnv.executeSql("CREATE TABLE dim_base_community(\n" +
                " id INT,\n" +
                " communityName STRING,\n" +
                " district_id INT,\n" +
                " city_id INT,\n" +
                " communityLocation STRING,\n" +
                " remark STRING,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_community',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive'\n" +
                ")\n");
        //7.读取地区维度表
        tEnv.executeSql("CREATE TABLE `dim_base_district` (\n" +
                " `id` INT,\n" +
                " `pid` INT,\n" +
                " `code` STRING,\n" +
                " `name` STRING,\n" +
                " `remark` STRING,\n" +
                " `create_time` TIMESTAMP,\n" +
                " `update_time` TIMESTAMP,\n" +
                " `status` SMALLINT,\n" +
                " `level` SMALLINT,\n" +
                " PRIMARY KEY (`id`)  NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = 'jdbc:mysql://master:3306/sca?useUnicode=true&characterEncoding=utf8',\n" +
                " 'table-name' = 'dim_base_district',\n" +
                " 'username' = 'hive',\n" +
                " 'password' = 'hive' \n" +
                ")\n");
        //8.窗口计算
        Table windowTable = tEnv.sqlQuery(
                "select  "+
                        "TUMBLE_START(time_ltz,INTERVAL '1' second) as stt,"+
                        "TUMBLE_END(time_ltz,INTERVAL '1' second) as edt,"+
                        "CURRENT_TIMESTAMP as ts,"+
                        "cId,"+
                        "owerId,"+
                        "carCode,"+
                        "sum(if(ctype=1,1,0)) car_in_cnt,"+
                        "sum(if(ctype=2,1,0)) car_out_cnt "+
                        " from dwd_cars_log GROUP BY TUMBLE(time_ltz,INTERVAL '1' second),cId,owerId,carCode "
        );
        //windowTable.execute().print();
        tEnv.createTemporaryView("windowTable",windowTable);

        //9.维度关联
        Table joinTable = tEnv.sqlQuery(
                "select "+
                        "car.carCode,"+
                        "car.carColor,"+
                        "car.type,"+
                        "ol.id as owerId,"+
                        "ol.owerSex,"+
                        "ol.owerAge,"+
                        "ol.owerHomeplace,"+
                        "ol.owerWorkplace,"+
                        "commu.id community_id,"+
                        "commu.communityName,"+
                        "commu.city_id,"+
                        "commu.district_id,"+
                        "distri.name district_name,"+
                        "cl.car_in_cnt,"+
                        "cl.car_out_cnt,"+
                        "cl.stt,"+
                        "cl.edt,"+
                        "cl.ts "+
                        "from windowTable cl "+
                        "join dim_base_ower ol on cl.owerId=ol.id "+
                        "join dim_base_cars car on cl.carCode=car.carCode "+
                        "join dim_base_community commu on cl.cId=commu.id "+
                        "join dim_base_district distri on commu.district_id=distri.id "
        );
        tEnv.createTemporaryView("joinTable",joinTable);

        DataStream<CarBehavior> carBehaviorDataStream=tEnv.toDataStream(joinTable, CarBehavior.class);

        String sql ="insert into default.dws_car_behavior (carCode,carColor,type,owerId,owerSex,owerAge," +
                "owerHomeplace,owerWorkplace,community_id,communityName,city_id,district_id,district_name," +
                "car_in_cnt,car_out_cnt,stt,edt,ts) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
        DwsCarBehaviorClickhouseSink jdbcSink = new DwsCarBehaviorClickhouseSink(sql);
        carBehaviorDataStream.addSink(jdbcSink);
        senv.execute("DwsCarBehavior");
    }
}
